package cep;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * next 严格模式
 */
public class TestCep05 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> ds = env.readTextFile("data/cep_04.txt");

        Pattern<String, String> pattern = Pattern.<String>begin("start").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value.startsWith("a");
            }
        })
                //匹配 a b  a后面不能跟c
                .notNext("notNext").where(new SimpleCondition<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        return value.startsWith("c");
                    }
                })
//                //匹配 a b  之间不能有c
//               .notFollowedBy("notFollowedBy").where(new SimpleCondition<String>() {
//                    @Override
//                    public boolean filter(String value) throws Exception {
//                        return value.startsWith("c");
//                    }
//                })
                .followedByAny("next").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value.startsWith("b");
            }
        });

        PatternStream<String> ds1 = CEP.pattern(ds, pattern);

        //a
        //b1
        //a
        //b2
        //匹配到 一条 赛选一次
        ds1.flatSelect(new PatternFlatSelectFunction<String, String>() {
            @Override
            public void flatSelect(Map<String, List<String>> map, Collector<String> collector) throws Exception {
                map.get("start").forEach(str->collector.collect(str));
                map.get("next").forEach(str->collector.collect(str));
            }
        }).print();

        env.execute();
    }
}
